package dm.service;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;

import org.dom4j.DocumentException;

import algo.LoadBalance;
import dm.bsi.Container;
import dm.bsi.Deployment;
import dm.bsi.Image;
import dm.bsi.Log;
import dm.bsi.MSDeployment;
import dm.bsi.Node;
import dm.bsi.Service;
import indi.zjh.dp.db.Configuration;
import indi.zjh.dp.db.Session;
import indi.zjh.dp.error.CommandExecuteException;
import indi.zjh.dp.error.ConfigureInvalidException;
import indi.zjh.dp.error.UnsuportedFieldException;
import indi.zjh.dp.util.RedisUtil;
import indi.zjh.dp.util.SystemCmd;
import javassist.tools.reflect.Sample;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisSentinelManager implements Manager {
	
	private class PingException extends Exception
	{
		public PingException()
		{
			super();
		}
		
		public PingException(String message)
		{
			super(message);
		}
	}
	
	private class ContainerState
	{
		public static final int STATE_INIT = 0;
		public static final int STATE_CONNECT = 1;
		public static final int STATE_PING = 2;
		
		public static final int MAX_TRY_COUNT = 3;
		
		public int state = STATE_INIT;
		public int tryCount = 0;
		private JedisPool jedisPool = null;
		private Jedis redis = null;
		private PrintWriter out = null;
		private BufferedReader in = null;
		private String host;
		private int port;
		
		public void connect(String host, int port) throws UnknownHostException, IOException
		{
			
			this.host = host;
			this.port = port;
			
			this.redis = RedisUtil.getJedis();
			
        	JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(1);
            config.setMaxIdle(1);
            config.setMaxWaitMillis(10000);
            config.setTestOnBorrow(true);

            if(this.redis != null)
            {
            	this.redis.close();
            }
            if(this.jedisPool != null)
            {
            	this.jedisPool.close();
            }
            
            this.jedisPool = new JedisPool(config, this.host, this.port, 10000);
	        this.redis = jedisPool.getResource();
			
		}
		
		public void ping() throws PingException
		{
			try
			{
				
				String ret = this.redis.ping().trim();
				if(!ret.equalsIgnoreCase("PONG"))
				{
					throw new Exception("ping异常");
				}
				this.tryCount = 0;
				
			}
			catch(Exception e)
			{
				throw new PingException(e.getMessage());
			}
		}
	}
	
	private static Hashtable<String, ContainerState> containers = new Hashtable<String, ContainerState>();


	@Override
	public void process(Session session, Deployment deployment, Service service) throws Exception {
		// TODO Auto-generated method stub
		try{
		MSDeployment msDeployment = (MSDeployment)deployment;
		
		long[] containerIds = service.getContainers().getIds();
		Container masterContainer = null;
		ArrayList<Container> slaveContainers = new ArrayList<Container>();
		for(int i = 0; i < containerIds.length; ++i)
		{
			Container container = (Container)session.getById(Container.class, containerIds[i]);
			
			try
			{
				this.ping(session, deployment, container);
				if(container.getRole() == Container.ROLE_MASTER)
				{
					masterContainer = container;
				}
				else if(container.getRole() == Container.ROLE_SLAVE)
				{
					slaveContainers.add(container);
				}
			}
			catch(Exception e)
			{
				container.destory(session, service);
				Log.info(session, deployment, "删除没有运行的节点, "+container.getNode(session).getIp()+":"+container.getDockerPort());
			}
			
		}


		// 获取准备就绪的服务器节点
		Node[] nodes = Node.listReadyNodes(session);
		if(nodes.length == 0)
		{
			Log.failure(session, deployment, "当前没有准备就绪的服务器节点, 部署容器被中断");
			return;
		}
		
		Image image = deployment.getImage(session);


		// 如果没有master, 启动master
		if(masterContainer == null)
		{
			if(slaveContainers.size() == 0)
			{
				Node selectedNode = LoadBalance.getIdlestNode(nodes);
				selectedNode.downloadImage(session, image);
				Container newContainer = new Container();
				HashMap<String, String> env = new HashMap<String, String>();
				env.put("ROLE", "MASTER");
				newContainer.setRole(Container.ROLE_MASTER);
				newContainer.start(session, deployment, selectedNode, image, true, env);
				Service.agentUpdate(session);
				Log.info(session, deployment, "启动master节点, "+selectedNode.getIp()+":"+newContainer.getDockerPort());
				return;
			}
			// 选一个slave当master
			else
			{
				// 选第一个节点当做master
				this.changeMaster(session, slaveContainers.get(0));
				
				Log.info(session, deployment, "选举新的master节点, "+slaveContainers.get(0).getNode(session).getIp()+":"+slaveContainers.get(0).getDockerPort());
				
				// 其余节点更新主节点信息
				for(int i = 1; i < slaveContainers.size(); ++i)
				{
					this.changeSlave(session, slaveContainers.get(i), slaveContainers.get(0));
					Log.info(session, deployment, "slave节点同步新的主节点, "+slaveContainers.get(i).getNode(session).getIp()+":"+slaveContainers.get(i).getDockerPort()+" => "+slaveContainers.get(0).getNode(session).getIp()+":"+slaveContainers.get(0).getDockerPort());
				}
				Service.agentUpdate(session);
				return;
			}
		}

		// slave节点太少, 启动slave
		if(slaveContainers.size() < msDeployment.getExpectedSlaveCount())
		{
			Node selectedNode = LoadBalance.getIdlestNode(nodes);
			selectedNode.downloadImage(session, image);
			Container newContainer = new Container();
			HashMap<String, String> env = new HashMap<String, String>();
			env.put("ROLE", "SLAVE");
			env.put("MASTER_HOST", masterContainer.getNode(session).getIp());
			env.put("MASTER_PORT", masterContainer.getDockerPort() + "");
			newContainer.setRole(Container.ROLE_SLAVE);
			newContainer.start(session, deployment, selectedNode, image, true, env);
			Log.info(session, deployment, "启动slave节点, "+selectedNode.getIp()+":"+newContainer.getDockerPort());
			return;
		}
		
		// 再次检查master节点

		HashMap<String, String> masterParam = this.getAttribute(session, masterContainer);
		if(masterContainer == null || masterParam.get("role") == null || !masterParam.get("role").equals("master"))
		{
			Log.info(session, deployment, "master节点为空或者异常, 忽略后续异常检测");
			return;
		}


		// 检测异常节点, 并且停掉

		containerIds = service.getContainers().getIds();
		slaveContainers = new ArrayList<Container>();
		for(int i = 0; i < containerIds.length; ++i)
		{
			Container container = (Container)session.getById(Container.class, containerIds[i]);
			HashMap<String, String> param = this.getAttribute(session, container);
			if(param.get("role") == null)
			{
				container.destory(session, service);
				Log.info(session, deployment, "删除没有响应的节点, "+container.getNode(session).getIp()+":"+container.getDockerPort());
			}
			else if(param.get("role").equals("slave"))
			{
				if(!param.get("master_link_status").equals("up"))
				{
					if(deployment.getTryCount() == 3)
					{
						deployment.setTryCount(0);
						container.destory(session, service);
						Log.info(session, deployment, "slave节点没有连接上master节点, 删除, "+container.getNode(session).getIp()+":"+container.getDockerPort());
					}
					deployment.setTryCount(deployment.getTryCount() + 1);
					session.save(deployment).commit();
				}
				else 
				{
					slaveContainers.add(container);
				}
			}
		}
		
		
		// 停掉多余的slave节点
		if(slaveContainers.size() > msDeployment.getExpectedSlaveCount())
		{
			int stopCount = slaveContainers.size() - msDeployment.getExpectedSlaveCount();
			for(int i = 0; i < slaveContainers.size(); ++i)
			{
				if(i < stopCount)
				{
					slaveContainers.get(i).destory(session, service);
					Log.info(session, deployment, "删除多于预期的slave节点, "+slaveContainers.get(i).getNode(session).getIp()+":"+slaveContainers.get(i).getDockerPort());
					return;
				}
			}
		}
		
		
		
		}catch(Exception e)
		{
			StringWriter sw = new StringWriter();
			e.printStackTrace(new PrintWriter(sw));
			Log.failure(session, deployment, sw.toString());
			
		}
	}

	/**
	 * 获取一个节点的信息
	 * @param session
	 * @param container
	 * @return
	 * @throws UnsuportedFieldException
	 * @throws DocumentException
	 * @throws Exception
	 */
	private HashMap<String, String> getAttribute(Session session, Container container) throws UnsuportedFieldException, DocumentException, Exception
	{
		HashMap<String, String> param = new HashMap<String, String>();
		try
		{
			Node relNode = container.getNode(session);
			String cmd = "ssh root@"+relNode.getIp()+" docker exec -i "+container.getOid()+" redis-cli -p "+container.getContainerPort()+" info replication";
			String response = SystemCmd.execCMD(cmd, 10);
			String[] lines = response.split("\n");
			for(int i = 0; i < lines.length; ++i)
			{
				String[] parts = lines[i].split(":");
				if(parts.length == 2)
					param.put(parts[0].trim(), parts[1].trim());
			}
		}
		catch(CommandExecuteException e)
		{
			
		}
		
		return param;
	}
	
	private boolean changeMaster(Session session, Container container) throws UnsuportedFieldException, DocumentException, Exception
	{
		container.setRole(Container.ROLE_MASTER);
		session.save(container).commit();
		
		Node relNode = container.getNode(session);
		String cmd = "ssh root@"+relNode.getIp()+" docker exec -i "+container.getOid()+" redis-cli -p "+container.getContainerPort()+" SLAVEOF NO ONE";
		String response = SystemCmd.execCMD(cmd, 10).toLowerCase().trim();
		if(response.contains("ok"))
			return true;
		return false;
	}
	
	private boolean changeSlave(Session session, Container container, Container masterContainer) throws UnsuportedFieldException, DocumentException, Exception
	{
		container.setRole(Container.ROLE_SLAVE);
		session.save(container).commit();
		
		Node relNode = container.getNode(session);
		Node masterNode = masterContainer.getNode(session);
		String cmd = "ssh root@"+relNode.getIp()+" docker exec -i "+container.getOid()+" redis-cli -p "+container.getContainerPort()+" SLAVEOF "+masterNode.getIp()+" "+masterContainer.getDockerPort();
		String response = SystemCmd.execCMD(cmd, 10).toLowerCase().trim();
		if(response.contains("ok"))
			return true;
		return false;
	}

	@Override
	public void ping(Session session, Deployment deployment, Container container) throws Exception {
		// TODO Auto-generated method stub
		
		// 容器未被添加进管理池
		if(!containers.containsKey(container.getOid()))
		{
			ContainerState cState = new ContainerState();
			containers.put(container.getOid(), cState);
		}

		ContainerState thisState = containers.get(container.getOid());
		Node node = container.getNode(session);
				
		try
		{
			switch (thisState.state) {
			case ContainerState.STATE_INIT:
				thisState.state = ContainerState.STATE_CONNECT;
				break;
					
			case ContainerState.STATE_CONNECT:
				thisState.connect(node.getIp(), container.getDockerPort());
				thisState.state = ContainerState.STATE_PING;
				break;
			case ContainerState.STATE_PING:
				thisState.ping();
				thisState.state = ContainerState.STATE_PING;
				break;
						
			default:
						
				break;
			}
		}
		catch(PingException ee)
		{
			thisState.state = ContainerState.STATE_INIT;
			Log.failure(session, deployment, "容器"+container.getOid()+"连接异常,"+ee.getMessage());
			if(thisState.tryCount++ >= ContainerState.MAX_TRY_COUNT)
			{
				containers.remove(container.getOid());
				throw new Exception(ee.getMessage());
			}
		}
	}
}
